package com.epic.engine.streams;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.Serializable;

public class StreamSource implements Serializable {
    private String topic;
    private String broker;
    public KafkaSource<String> createSource(){
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(broker)
                .setTopics("aa","bbb")
                .setGroupId("test")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        return source;
    }
}
/**
 * flink 1.14 提供了KafkaSource and KafkaSink 来跟踪kafka的版本
 * https://www.136.la/nginx/show-221487.html
 */
